package com.atguigu.flink.chapter11;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/12/21 10:04
 */
public class Flink06_SQL_File {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        
        // 建立一个表, 与文件关联
        tEnv.executeSql("create table sensor(" +
                            "   id string, " +
                            "   ts bigint, " +
                            "   vc int" +
                            ")with(" +
                            "   'connector'='filesystem', " +
                            "   'path'='input/sensor.txt', " +
                            "   'format'='csv' " +
                            ")");
    
        Table table = tEnv.sqlQuery("select * from sensor where id='sensor_1'");
    
        // 动态表, 与sink关联
        tEnv.executeSql("create table `out`(" +
                            "   id string, " +
                            "   ts bigint, " +
                            "   vc int" +
                            ")with(" +
                            "   'connector'='filesystem', " +
                            "   'path'='input/a.txt', " +
                            "   'format'='csv' " +
                            ")");
    
        table.executeInsert("out");
    
    }
}
